//! 客户端集成测试

use nacos_client::{NacosClient, ClientConfig, ServiceInstance, ConfigListener, ServiceListener};
use std::sync::Arc;
use std::collections::HashMap;
use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

/// 测试配置监听器
struct TestConfigListener {
    tx: mpsc::UnboundedSender<String>,
}

#[async_trait::async_trait]
impl ConfigListener for TestConfigListener {
    async fn on_config_changed(&self, data_id: &str, group: &str, content: &str) {
        self.tx.send(format!("changed:{}:{}:{}", data_id, group, content)).unwrap();
    }

    async fn on_config_added(&self, data_id: &str, group: &str, content: &str) {
        self.tx.send(format!("added:{}:{}:{}", data_id, group, content)).unwrap();
    }

    async fn on_config_removed(&self, data_id: &str, group: &str) {
        self.tx.send(format!("removed:{}:{}", data_id, group)).unwrap();
    }
}

/// 测试服务监听器
struct TestServiceListener {
    tx: mpsc::UnboundedSender<String>,
}

#[async_trait::async_trait]
impl ServiceListener for TestServiceListener {
    async fn on_service_changed(&self, service_name: &str, instances: &[ServiceInstance]
    ) {
        self.tx.send(format!("service_changed:{}:count={}", service_name, instances.len())).unwrap();
    }

    async fn on_instance_added(&self, service_name: &str, instance: &ServiceInstance) {
        self.tx.send(format!("instance_added:{}:{}:{}", service_name, instance.ip, instance.port)).unwrap();
    }

    async fn on_instance_removed(&self, service_name: &str, instance: &ServiceInstance) {
        self.tx.send(format!("instance_removed:{}:{}:{}", service_name, instance.ip, instance.port)).unwrap();
    }
}

#[tokio::test]
async fn test_client_creation() {
    let config = ClientConfig::new("127.0.0.1:8848");
    let client = NacosClient::new(config).await;
    
    assert!(client.is_ok());
}

#[tokio::test]
async fn test_naming_service_lifecycle() {
    let config = ClientConfig::new("127.0.0.1:8848");
    let client = Arc::new(NacosClient::new(config).await.unwrap());
    let naming_client = client.naming_client();

    let service_name = "test-service";
    let instance = ServiceInstance {
        ip: "127.0.0.1".to_string(),
        port: 8080,
        weight: 1.0,
        healthy: true,
        enabled: true,
        ephemeral: true,
        cluster_name: "default".to_string(),
        service_name: service_name.to_string(),
        namespace: "public".to_string(),
        metadata: HashMap::new(),
    };

    // 注册服务
    let result = naming_client.register_instance(service_name, instance.clone()).await;
    assert!(result.is_ok());

    // 等待服务注册
    sleep(Duration::from_millis(100)).await;

    // 发现服务
    let instances = naming_client.get_instances(service_name, true).await;
    assert!(instances.is_ok());
    assert!(!instances.unwrap().is_empty());

    // 注销服务
    let result = naming_client.deregister_instance(service_name, "127.0.0.1", 8080).await;
    assert!(result.is_ok());
}

#[tokio::test]
async fn test_config_service_lifecycle() {
    let config = ClientConfig::new("127.0.0.1:8848");
    let client = Arc::new(NacosClient::new(config).await.unwrap());
    let config_client = client.config_client("public", "DEFAULT_GROUP");

    let data_id = "test-config";
    let group = "DEFAULT_GROUP";
    let content = "test-content";

    // 发布配置
    let result = config_client.publish_config(data_id, group, content).await;
    assert!(result.is_ok());
    assert!(result.unwrap());

    // 等待配置发布
    sleep(Duration::from_millis(100)).await;

    // 获取配置
    let retrieved_content = config_client.get_config(data_id, group).await;
    assert!(retrieved_content.is_ok());
    assert_eq!(retrieved_content.unwrap(), content);

    // 删除配置
    let result = config_client.remove_config(data_id, group).await;
    assert!(result.is_ok());
    assert!(result.unwrap());
}

#[tokio::test]
async fn test_config_listener() {
    let config = ClientConfig::new("127.0.0.1:8848");
    let client = Arc::new(NacosClient::new(config).await.unwrap());
    let config_client = client.config_client("public", "DEFAULT_GROUP");

    let (tx, mut rx) = mpsc::unbounded_channel();
    let listener = Box::new(TestConfigListener { tx });

    let data_id = "test-listener-config";
    let group = "DEFAULT_GROUP";

    // 添加监听器
    let result = config_client.add_listener(data_id, group, listener).await;
    assert!(result.is_ok());

    // 发布配置
    let result = config_client.publish_config(data_id, group, "initial-content").await;
    assert!(result.is_ok());

    // 等待监听事件
    sleep(Duration::from_millis(200)).await;

    // 更新配置
    let result = config_client.publish_config(data_id, group, "updated-content").await;
    assert!(result.is_ok());

    // 验证监听器收到事件
    let mut events = Vec::new();
    while let Ok(Some(event)) = tokio::time::timeout(Duration::from_millis(500), rx.recv()).await {
        events.push(event);
    }

    assert!(!events.is_empty());
    assert!(events.iter().any(|e| e.contains("added")));
    assert!(events.iter().any(|e| e.contains("changed")));

    // 清理
    let _ = config_client.remove_config(data_id, group).await;
}

#[tokio::test]
async fn test_service_listener() {
    let config = ClientConfig::new("127.0.0.1:8848");
    let client = Arc::new(NacosClient::new(config).await.unwrap());
    let naming_client = client.naming_client();

    let (tx, mut rx) = mpsc::unbounded_channel();
    let listener = Box::new(TestServiceListener { tx });

    let service_name = "test-listener-service";

    // 添加监听器
    let result = naming_client.subscribe_service(service_name, listener).await;
    assert!(result.is_ok());

    // 注册服务
    let instance = ServiceInstance {
        ip: "127.0.0.1".to_string(),
        port: 8081,
        weight: 1.0,
        healthy: true,
        enabled: true,
        ephemeral: true,
        cluster_name: "default".to_string(),
        service_name: service_name.to_string(),
        namespace: "public".to_string(),
        metadata: HashMap::new(),
    };

    let result = naming_client.register_instance(service_name, instance).await;
    assert!(result.is_ok());

    // 等待监听事件
    sleep(Duration::from_millis(200)).await;

    // 注销服务
    let result = naming_client.deregister_instance(service_name, "127.0.0.1", 8081).await;
    assert!(result.is_ok());

    // 验证监听器收到事件
    let mut events = Vec::new();
    while let Ok(Some(event)) = tokio::time::timeout(Duration::from_millis(500), rx.recv()).await {
        events.push(event);
    }

    assert!(!events.is_empty());
    assert!(events.iter().any(|e| e.contains("instance_added")));
    assert!(events.iter().any(|e| e.contains("instance_removed")));
}

#[tokio::test]
async fn test_cache_functionality() {
    let config = ClientConfig::new("127.0.0.1:8848");
    let client = Arc::new(NacosClient::new(config).await.unwrap());
    
    // 测试配置缓存
    let config_client = client.config_client("public", "test-cache-group");
    let data_id = "test-cache-config";
    let content = "cache-test-content";

    // 第一次获取，应该缓存
    let result = config_client.publish_config(data_id, "test-cache-group", content).await;
    assert!(result.is_ok());

    sleep(Duration::from_millis(100)).await;

    let cached_content = config_client.get_config_cached(data_id, "test-cache-group").await;
    assert!(cached_content.is_ok());
    assert_eq!(cached_content.unwrap(), content);

    // 测试命名缓存
    let naming_client = client.naming_client();
    let service_name = "test-cache-service";
    let instance = ServiceInstance {
        ip: "127.0.0.1".to_string(),
        port: 8082,
        weight: 1.0,
        healthy: true,
        enabled: true,
        ephemeral: true,
        cluster_name: "default".to_string(),
        service_name: service_name.to_string(),
        namespace: "public".to_string(),
        metadata: HashMap::new(),
    };

    let result = naming_client.register_instance(service_name, instance).await;
    assert!(result.is_ok());

    sleep(Duration::from_millis(100)).await;

    let cached_instances = naming_client.get_instances_cached(service_name, true).await;
    assert!(cached_instances.is_ok());
    assert!(!cached_instances.unwrap().is_empty());

    // 清理
    let _ = config_client.remove_config(data_id, "test-cache-group").await;
    let _ = naming_client.deregister_instance(service_name, "127.0.0.1", 8082).await;
}

#[tokio::test]
async fn test_batch_operations() {
    let config = ClientConfig::new("127.0.0.1:8848");
    let client = Arc::new(NacosClient::new(config).await.unwrap());
    let config_client = client.config_client("public", "test-batch-group");

    let data_ids = vec!["batch-config-1", "batch-config-2", "batch-config-3"];
    let content = "batch-test-content";

    // 批量发布配置
    for data_id in &data_ids {
        let result = config_client.publish_config(data_id, "test-batch-group", content).await;
        assert!(result.is_ok());
    }

    sleep(Duration::from_millis(200)).await;

    // 批量获取配置
    let results = config_client.batch_get_config(&data_ids.iter().map(|s| *s).collect::<Vec<&str>>(),
        "test-batch-group"
    ).await;
    
    assert!(results.is_ok());
    let results = results.unwrap();
    assert_eq!(results.len(), data_ids.len());

    // 验证所有配置都获取到
    for data_id in &data_ids {
        assert_eq!(results.get(*data_id).unwrap(), content);
    }

    // 清理
    for data_id in &data_ids {
        let _ = config_client.remove_config(data_id, "test-batch-group").await;
    }
}

#[tokio::test]
async fn test_health_check() {
    let config = ClientConfig::new("127.0.0.1:8848");
    let client = NacosClient::new(config).await;
    
    if let Ok(client) = client {
        let health = client.health_check().await;
        if let Ok(healthy) = health {
            assert!(healthy);
        }
    }
}

#[tokio::test]
async fn test_retry_mechanism() {
    // 测试重试机制
    let config = ClientConfig::new("invalid-server:8848")
        .with_retry(3, 100);
    let client = NacosClient::new(config).await;
    
    // 应该失败但有重试
    assert!(client.is_err());
}

#[tokio::test]
async fn test_concurrent_operations() {
    let config = ClientConfig::new("127.0.0.1:8848");
    let client = Arc::new(NacosClient::new(config).await.unwrap());
    
    let mut handles = vec![];
    
    // 并发服务注册
    for i in 0..5 {
        let client = client.clone();
        let handle = tokio::spawn(async move {
            let naming_client = client.naming_client();
            let service_name = format!("concurrent-service-{}", i);
            let instance = ServiceInstance {
                ip: "127.0.0.1".to_string(),
                port: 8000 + i,
                weight: 1.0,
                healthy: true,
                enabled: true,
                ephemeral: true,
                cluster_name: "default".to_string(),
                service_name: service_name.clone(),
                namespace: "public".to_string(),
                metadata: HashMap::new(),
            };

            naming_client.register_instance(&service_name, instance
            ).await
        });
        handles.push(handle);
    }

    // 并发配置操作
    for i in 0..5 {
        let client = client.clone();
        let handle = tokio::spawn(async move {
            let config_client = client.config_client("public", "concurrent-group");
            let data_id = format!("concurrent-config-{}", i);
            
            config_client.publish_config(
                &data_id,
                "concurrent-group",
                "concurrent-content"
            ).await
        });
        handles.push(handle);
    }

    // 等待所有操作完成
    for handle in handles {
        let result = handle.await;
        assert!(result.is_ok());
    }

    // 验证所有服务都存在
    let naming_client = client.naming_client();
    for i in 0..5 {
        let service_name = format!("concurrent-service-{}", i);
        let instances = naming_client.get_instances(&service_name, true
        ).await;
        assert!(instances.is_ok());
    }

    // 清理
    for i in 0..5 {
        let service_name = format!("concurrent-service-{}", i);
        let _ = naming_client.deregister_instance(&service_name, "127.0.0.1", 8000 + i
        ).await;
        
        let config_client = client.config_client("public", "concurrent-group");
        let data_id = format!("concurrent-config-{}", i);
        let _ = config_client.remove_config(&data_id, "concurrent-group"
        ).await;
    }
}